/**
 * AbstractServer.java 2020/5/13 4:43 下午 Copyright ©2020 www.bmsoft.com.cn All rights reserved.
 * PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */
package remoting;

import common.constants.CommonConstants;
import common.exception.RemotingException;
import common.model.Callback;
import common.model.Codec2;
import common.model.Endpoint;
import common.model.MagicUrl;
import common.model.Resetable;
import common.util.ExecutorUtil;
import common.util.NetUtils;
import config.MagicReferenceConfig;
import config.MagicServiceConfig;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import remoting.channel.Channel;
import remoting.channel.ChannelHandler;
import serialization.Serializer;

import static common.constants.CommonConstants.ANYHOST_KEY;
import static common.constants.CommonConstants.ANYHOST_VALUE;
import static common.constants.CommonConstants.THREADS_KEY;
import static common.constants.CommonConstants.IDLE_TIMEOUT_KEY;
import static common.constants.CommonConstants.DEFAULT_IDLE_TIMEOUT;
import static common.constants.CommonConstants.ACCEPTS_KEY;
import static common.constants.CommonConstants.DEFAULT_ACCEPTS;

/**
 * File：AbstractServer.java<br>
 * Title: <br>
 * Description: <br>
 * Company: www.treefinance.com.cn <br>
 * @author heyouchi
 * Date: 2020/5/13
 */
public abstract class AbstractServer extends AbstractPeer implements Server, Resetable {

  protected static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);
  protected static final String SERVER_THREAD_POOL_NAME = "MagicServerHandler";
  ExecutorService executor;
  private InetSocketAddress localAddress;
  private InetSocketAddress bindAddress;
  private int accepts;
  private int idleTimeout;

  public AbstractServer(MagicUrl url, ChannelHandler channelHandler, Codec2 codec)
      throws RemotingException {
    super(url, channelHandler, codec);

    localAddress = getUrl().toInetSocketAddress();

    String bindIp = getUrl().getParameter(CommonConstants.BIND_IP_KEY, getUrl().getHost());
    int bindPort = getUrl().getParameter(CommonConstants.BIND_PORT_KEY, getUrl().getPort());
    if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
      bindIp = ANYHOST_VALUE;
    }
    bindAddress = new InetSocketAddress(bindIp, bindPort);
    this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
    this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT);
    try {
      doOpen();
      if (logger.isInfoEnabled()) {
        logger.info(
            "Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export "
                + getLocalAddress());
      }
    } catch (Throwable t) {
      throw new RemotingException(url.toInetSocketAddress(), null,
          "Failed to bind " + getClass().getSimpleName()
              + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
    }
  }

  protected abstract void doOpen() throws Throwable;

  protected abstract void doClose() throws Throwable;

  @Override
  public void reset(MagicUrl url) {
    if (url == null) {
      return;
    }
    try {
      if (url.hasParameter(ACCEPTS_KEY)) {
        int a = url.getParameter(ACCEPTS_KEY, 0);
        if (a > 0) {
          this.accepts = a;
        }
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    try {
      if (url.hasParameter(IDLE_TIMEOUT_KEY)) {
        int t = url.getParameter(IDLE_TIMEOUT_KEY, 0);
        if (t > 0) {
          this.idleTimeout = t;
        }
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    try {
      if (url.hasParameter(THREADS_KEY)
          && executor instanceof ThreadPoolExecutor && !executor.isShutdown()) {
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;
        int threads = url.getParameter(THREADS_KEY, 0);
        int max = threadPoolExecutor.getMaximumPoolSize();
        int core = threadPoolExecutor.getCorePoolSize();
        if (threads > 0 && (threads != max || threads != core)) {
          if (threads < core) {
            threadPoolExecutor.setCorePoolSize(threads);
            if (core == max) {
              threadPoolExecutor.setMaximumPoolSize(threads);
            }
          } else {
            threadPoolExecutor.setMaximumPoolSize(threads);
            if (core == max) {
              threadPoolExecutor.setCorePoolSize(threads);
            }
          }
        }
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    super.setUrl(getUrl().addParameters(url.getParameters()));
  }

  @Override
  public void sendMessage(Object message, boolean sent) throws RemotingException {
    Collection<Channel> channels = getChannels();
    for (Channel channel : channels) {
      if (channel.isConnected()) {
        channel.sendMessage(message, sent);
      }
    }
  }

  @Override
  public void close() {
    if (logger.isInfoEnabled()) {
      logger.info("Close " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export "
          + getLocalAddress());
    }
    ExecutorUtil.shutdownNow(executor, 100);
    try {
      super.close();
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }
    try {
      doClose();
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }
  }

  @Override
  public void close(int timeout) {
    ExecutorUtil.gracefulShutdown(executor, timeout);
    close();
  }

  @Override
  public InetSocketAddress getLocalAddress() {
    return localAddress;
  }

  public InetSocketAddress getBindAddress() {
    return bindAddress;
  }

  public int getAccepts() {
    return accepts;
  }

  public int getIdleTimeout() {
    return idleTimeout;
  }

  @Override
  public void connected(Channel ch) throws RemotingException {
    // If the server has entered the shutdown process, reject any new connection
    if (this.isClosing() || this.isClosed()) {
      logger.warn("Close new channel " + ch
          + ", cause: server is closing or has been closed. For example, receive a new connect request while in shutdown process.");
      ch.close();
      return;
    }

    Collection<Channel> channels = getChannels();
    if (accepts > 0 && channels.size() > accepts) {
      logger.error("Close channel " + ch + ", cause: The server " + ch.getLocalAddress()
          + " connections greater than max config " + accepts);
      ch.close();
      return;
    }
    super.connected(ch);
  }

  @Override
  public void disconnected(Channel ch) throws RemotingException {
    Collection<Channel> channels = getChannels();
    if (channels.isEmpty()) {
      logger.warn("All clients has disconnected from " + ch.getLocalAddress()
          + ". You can graceful shutdown now.");
    }
    super.disconnected(ch);
  }

}
